#include "postgre_database.h"
#include "exceptions.h"
#include "watch.h"
#include "sql.h"
#include "str.h"

void postgre_database::check_status(PGresult *result, ExecStatusType type, bool clear_result /*= true*/, bool end_transaction_on_failure /*= true*/) {
  if (PQresultStatus(result) != type) { // there is an error
    const string err = PQerrorMessage(connection);

    PQclear(result); // clear the result in case of error

    if (is_in_transaction == true)
      rollback();
    else if (end_transaction_on_failure == true) {
      result = PQexec(connection, "END");
      PQclear(result);
    } // if

    throw db_exception(err);
  } else if (clear_result == true) // there is no error
    PQclear(result);
} // postgre_database::check_status

void postgre_database::connect(const string &connection_string, const string &protocol_file /*= global_empty_string*/) {
  watch w(true);
  if (connected == true)
    disconnect();

  protocol = protocol_file;

  // make a connection to the database
  connection = PQconnectdb(connection_string.c_str());

  // check to see that the backend connection was successfully made
  if (PQstatus(connection) != CONNECTION_OK) {
    const string msg = PQerrorMessage(connection);

    PQfinish(connection);
    connection = 0;

    throw db_exception(msg);
  } // if

  connected = true;
  prot("CON", "postgre_database", w.get_elapsed_milliseconds());
} // postgre_database::connect

void postgre_database::connect(const string &host, const string &user_id, const string &password, const string &db_name, const string &protocol_file /*= global_empty_string*/) {
  string format = "hostaddr=%1 dbname=%2 user=%3 password=%4";
  str::format(format, arg<string>(host, db_name, user_id, password));
  connect(format, protocol_file);
} // postgre_database::connect

void postgre_database::disconnect() {
  // close the connection to the database and cleanup
  if (connected == true) {
    watch w(true);

    rollback();

    PQfinish(connection);
    connection = 0;

    connected = false;

    prot("DCO", "postgre_database", w.get_elapsed_milliseconds());
  } // if
} // postgre_database::disconnect

bool postgre_database::is_connected() {
  if (connected == false)
    return false;

  PGresult *result = PQexec(connection, "SELECT VERSION();");
  if (PQresultStatus(result) != PGRES_TUPLES_OK)
    connected = false;

  PQclear(result);
  return connected;
} // postgre_database::is_connected

void postgre_database::execute_query(const string &s, record_set &rs, bool append_to_record_set) {
  watch w(true);
  if (connected == false)
    throw db_exception(PQerrorMessage(connection));

  PGresult *result = PQexec(connection, s.c_str());
  check_status(result, PGRES_TUPLES_OK, false, false); // if everything works ok, then the 'result' must not be cleared here

  const unsigned int n_fields = PQnfields(result); // number of columns returned
  const unsigned int n_rows = PQntuples(result); // number of rows returned

  if (append_to_record_set == false) {
    rs.reconstruct(n_rows, n_fields);

    // fills the record_set with column names
    for (unsigned int i = 0; i < n_fields; ++i) {
      const char *col_name = PQfname(result, i);
      rs.fill_record_set(col_name, strlen(col_name));
    } // for i
  } else if (rs.get_column_count() == n_fields)
    rs.reserve(rs.get_row_count() + n_rows, n_fields);
  else
    throw db_exception("Number of columns are not the same! (append_to_record_set = true)");

  // fills the record_set with data
  for (unsigned int i = 0; i < n_rows; ++i) {
    for (unsigned int j = 0; j < n_fields; ++j)
      rs.fill_record_set(PQgetvalue(result, i, j), PQgetlength(result, i, j)); // fills the record_set with data
  } // for i

  PQclear(result);

  prot("QRY", s, w.get_elapsed_milliseconds(), rs.get_row_count());
} // postgre_database::execute_query

void postgre_database::execute_update(const string &s) {
  watch w(true);
  if (connected == false)
    throw db_exception(PQerrorMessage(connection));

  PGresult *result = PQexec(connection, s.c_str());
  check_status(result, PGRES_COMMAND_OK, true, false);

  prot("UPD", s, w.get_elapsed_milliseconds());
} // postgre_database::execute_update

void postgre_database::start_transaction() {
  if (is_in_transaction == false) {
    watch w(true);
    is_in_transaction = true;
    PGresult *result = PQexec(connection, "BEGIN;");
    check_status(result, PGRES_COMMAND_OK, true, true);
    prot("TRA", "STARTED", w.get_elapsed_milliseconds());
  } // if
} // postgre_database::start_transaction

void postgre_database::commit() {
  if (is_in_transaction == true) {
    watch w(true);
    is_in_transaction = false;
    PGresult *result = PQexec(connection, "COMMIT;");
    PQclear(result);
    prot("TRA", "COMMIT", w.get_elapsed_milliseconds());
  } // if
} // postgre_database::commit

void postgre_database::rollback() {
  if (is_in_transaction == true) {
    watch w(true);
    is_in_transaction = false;
    PGresult *result = PQexec(connection, "ROLLBACK;");
    PQclear(result);
    prot("TRA", "ROLLBACK", w.get_elapsed_milliseconds());
  } // if
} // postgre_database::rollback

/*
// transaction dependent implementation, which may conflict with existing transactions
void postgre_database::lock_table(const string &table_name) {
  watch w(true);
  if (table_name.empty() == true)
    throw parameter_exception("table_name parameter is required to lock the table!");

  string s = "BEGIN; LOCK %1;"; // default: ACCESS EXCLUSIVE
  str::format(s, arg<string>(table_name));
  execute_update(s);
  s = "TABLE %1 LOCKED";
  str::format(s, arg<string>(table_name));
  prot("LCK", s, w.get_elapsed_milliseconds());
} // postgre_database::lock_table

void postgre_database::unlock_table(const string &table_name) {
  watch w(true);
  execute_update("COMMIT;");
  string s = "TABLE %1 UNLOCKED";
  str::format(s, arg<string>(table_name));
  prot("LCK", s, w.get_elapsed_milliseconds());
} // postgre_database::unlock_table
*/

void postgre_database::lock_table(const string &table_name) {
  watch w(true);
  if (table_name.empty() == true)
    throw parameter_exception("table_name parameter is required to lock the table!");

  string s = "SELECT pg_advisory_lock(id) FROM %1 LIMIT 1;"; // table must have an id column
  str::format(s, arg<string>(table_name));

  record_set rs;
  execute_query(s, rs);
  s = "TABLE %1 LOCKED";
  str::format(s, arg<string>(table_name));
  prot("LCK", s, w.get_elapsed_milliseconds());
} // postgre_database::lock_table

void postgre_database::unlock_table(const string &table_name) {
  watch w(true);
  record_set rs;
  execute_query("SELECT pg_advisory_unlock_all();", rs);
  string s = "TABLE %1 UNLOCKED";
  str::format(s, arg<string>(table_name));
  prot("LCK", s, w.get_elapsed_milliseconds());
} // postgre_database::unlock_table

postgre_database::postgre_database() : connection(0) {
  // nothing to do
} // postgre_database::postgre_database

postgre_database::~postgre_database() {
  disconnect();
} // postgre_database::~postgre_database
